import sys
import pendulum
from datetime import timedelta
import json
from typing import List, Union
from utils.macros import cst_hour,hour_add
import urllib.request
import pandas as pd
from pandas import DataFrame


class BaseElement:
    def __init__(self, tag: str):
        self.tag = tag


class HrElement(BaseElement):
    def __init__(self):
        tag = "hr"
        super().__init__(tag)


class TextElement(BaseElement):
    def __init__(self, content: str):
        super().__init__("div")
        self.text = {"content": content, "tag": "plain_text"}


class MarkdownElement(BaseElement):
    def __init__(self, content, text_align="center"):
        tag = "markdown"
        super().__init__(tag)
        self.text_align = text_align
        self.content = str(content)


class ColumnElement(BaseElement):
    def __init__(self, markdown_content: str, width="weighted", weight=1, vertical_align="top"):
        tag = "column"
        super().__init__(tag)
        self.width = width
        self.weight = weight
        self.vertical_align = vertical_align
        self.elements = [MarkdownElement(markdown_content)]


class ColumnSetElement(BaseElement):
    def __init__(self, columns_list=None, background_style="grey", flex_mode="none", horizontal_spacing="default"):
        tag = "column_set"
        super().__init__(tag)
        self.flex_mode = flex_mode
        self.background_style = background_style
        self.horizontal_spacing = horizontal_spacing
        if columns_list is None:
            columns_list = []
        self.columns = columns_list

    def append(self, element: ColumnElement):
        self.columns.append(element)


class Card:
    def __init__(self, header_content: str, elements: List[BaseElement], template="green"):
        self.elements = elements
        self.config = {"wide_screen_mode": True}
        self.header = {
            "template": template,
            "title": {
                "content": header_content,
                "tag": "plain_text"
            }
        }


class MarkdownCard:
    def __init__(self, card: Card):
        self.msg_type = "interactive"
        self.card = card


def get_table(df: DataFrame, data_column=None, column_weight: List[int] = None) -> List[
    BaseElement]:
    """
    这是一个将df转换为卡片的方法
    :param df: 数据
    :param data_column: 数据需要打印的列, 不填默认所有列
    :param high_light_column: 染色行使用的列。支持 green red grey default
    """
    column_set_list: List[BaseElement] = []
    if df.empty:
        return column_set_list
    # 打印列为空的情况取所有列
    if data_column is None:
        data_column: List[str] = df.columns.tolist()
    # 但是要去掉染色列
    for col in data_column:
        if "_color" in col:
            data_column.remove(col)
    if column_weight is None:
        column_weight = [1 for i in range(len(data_column))]
    column_set = ColumnSetElement()
    column_set_list.append(column_set)
    for idx, header in enumerate(data_column):
        column_set.append(ColumnElement(f"**{header}**", "weighted", column_weight[idx], "center"))
    for _, data in df.iterrows():
        column_set = ColumnSetElement()
        column_set_list.append(column_set)
        for idx, column in enumerate(data_column):
            color = data.get(column + "_color", default="default")
            markdown_content = f"""<font color='{color}'>{data[column]}</font>"""
            # if color == "red":
            #     markdown_content = f"**{markdown_content}**"
            column_set.append(ColumnElement(markdown_content, "weighted", column_weight[idx], "center"))
    return column_set_list


def object_to_dict(obj):
    if isinstance(obj, List) or isinstance(obj, tuple):
        return [object_to_dict(item) for item in obj]
    elif isinstance(obj, dict):
        return {key: object_to_dict(value) for key, value in obj.items()}
    elif hasattr(obj, "__dict__"):
        return object_to_dict(obj.__dict__)
    else:
        return obj


def send(url: str, card: Card, webhook: str):
    markdown_card = MarkdownCard(card=card)
    data = json.dumps(object_to_dict(markdown_card)).encode('utf-8')
    print(data)
    response = urllib.request.urlopen(
        urllib.request.Request(url=url, data=data, headers={"Content-Type": "application/json", "url": webhook})
    ).read()
    print(response)


def send_feishu(url: str, webhook: str, header: str, table: Union[DataFrame, List[BaseElement]] = None, content: str = None,
                column_weight: List[int] = None):
    """
    这是一个发送飞书卡片信息的方法
    :param webhook: 机器人webhook
    :param header: 卡片头
    :param content: 卡片详情信息
    :param table: 希望打印表格则传入 table
                  如果是 DataFrame 直接传入, 会读取除 color 列以外的列生成 table,
                  需要定制打印列和染色列 需要使用 get_table 将 DataFrame 转换为 List[BaseElement] 传入
                  get_table 转换可以选择打印的列和染色列
                  color 会默认作为染色行, 仅支持 green red grey default
                  有其余的定制需求需要自己创建 List[BaseElement] 传入
    """
    elements: List[BaseElement] = []
    if content is not None and content != '':
        elements.append(TextElement(content=content))
    # 无 table 的情况, 直接返回
    if table is None:
        return send(Card(header, elements))
    if isinstance(table, DataFrame):
        table = get_table(table, None, column_weight)
    if table is not None and len(table) != 0:
        elements.append(HrElement())
        elements.extend(table)
    return send(url, Card(header, elements), webhook)


def send_message(url:str, webhook: str, header: str, content: str, theme_color="green"):
    def _send_message(ctx):
        elements = []
        if content is not None and content != '':
            batch = cst_hour(hour_add(ctx["execution_date"], 6))
            elements.append(TextElement(content=f"【{batch}】批次{content}"))

        cst = pendulum.timezone('Asia/Shanghai')
        day = cst.convert(ctx['ti'].execution_date) + timedelta(hours=6)
        schedule_date = ['00']
        if day.strftime('%H') not in schedule_date:
            send(url, Card(header, elements, theme_color), webhook)
    return _send_message


def alert(url:str, webhook: str, header: str, jsonDf: str, colWeightListStr: str):
    send_feishu(url, webhook, header, pd.read_json(jsonDf), None, list(map(int, colWeightListStr.split(","))))





